package com.example.kafka.exception.consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.sql.SQLException;
import java.util.concurrent.TimeoutException;

/**
 * 处理消费者线程终止异常的Service
 */
@Service
public class ConsumerExceptionService {
    private final Logger log = LoggerFactory.getLogger(getClass());

    /**
     * 用例1: 处理反序列化异常
     * 模拟JSON解析失败的情况
     */
    @KafkaListener(topics = "deserialization-error-topic", groupId = "deserialization-error-group")
    public String handleDeserializationError(String message) {
        try {
            // 模拟JSON解析
            if (message.contains("invalid")) {
                throw new IOException("Invalid JSON format");
            }
            return "Successfully processed message: " + message;
        } catch (IOException e) {
            log.error("Deserialization error occurred: {}", e.getMessage());
            // 记录错误但不中断消费线程
            return "Error occurred but consumer continues: " + e.getMessage();
        }
    }

    /**
     * 用例2: 处理数据库连接异常
     * 模拟数据库操作失败的情况
     */
    @KafkaListener(topics = "db-connection-error-topic", groupId = "db-connection-error-group")
    public String handleDbConnectionError(String message) {
        try {
            // 模拟数据库操作
            if (message.contains("db-error")) {
                throw new SQLException("Database connection failed");
            }
            return "Successfully processed message: " + message;
        } catch (SQLException e) {
            log.error("Database error occurred: {}", e.getMessage());
            // 记录错误并继续处理
            return "Database error occurred but consumer continues: " + e.getMessage();
        }
    }

    /**
     * 用例3: 处理网络超时异常
     * 模拟外部服务调用超时的情况
     */
    @KafkaListener(topics = "network-timeout-topic", groupId = "network-timeout-group")
    public String handleNetworkTimeout(String message) {
        try {
            // 模拟外部服务调用
            if (message.contains("timeout")) {
                throw new TimeoutException("External service timeout");
            }
            return "Successfully processed message: " + message;
        } catch (TimeoutException e) {
            log.error("Network timeout occurred: {}", e.getMessage());
            // 记录错误并继续处理
            return "Timeout occurred but consumer continues: " + e.getMessage();
        }
    }

    /**
     * 用例4: 处理内存溢出异常
     * 模拟内存不足的情况
     */
    @KafkaListener(topics = "out-of-memory-topic", groupId = "out-of-memory-group")
    public String handleOutOfMemory(String message) {
        try {
            // 模拟内存密集型操作
            if (message.contains("large-data")) {
                throw new OutOfMemoryError("Insufficient memory");
            }
            return "Successfully processed message: " + message;
        } catch (OutOfMemoryError e) {
            log.error("Out of memory error occurred: {}", e.getMessage());
            // 记录错误并继续处理
            return "Memory error occurred but consumer continues: " + e.getMessage();
        }
    }

    /**
     * 用例5: 处理线程中断异常
     * 模拟线程被中断的情况
     */
    @KafkaListener(topics = "thread-interrupt-topic", groupId = "thread-interrupt-group")
    public String handleThreadInterrupt(String message) {
        try {
            // 模拟长时间运行的操作
            if (message.contains("interrupt")) {
                Thread.currentThread().interrupt();
                throw new InterruptedException("Thread interrupted");
            }
            return "Successfully processed message: " + message;
        } catch (InterruptedException e) {
            log.error("Thread interrupted: {}", e.getMessage());
            // 重新设置中断标志
            Thread.currentThread().interrupt();
            // 记录错误并继续处理
            return "Thread interrupted but consumer continues: " + e.getMessage();
        }
    }
}
